package cn.vetech.sync.elasticsearch.service.impl;

import cn.vetech.sync.elasticsearch.service.DataService;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import cn.vetech.sync.elasticsearch.service.ElasticsearchService;
import cn.vetech.sync.elasticsearch.util.JsonUtil;

import javax.annotation.Resource;
import java.util.Map;

/**
 * @author <a href="mailto:wangchao.star@gmail.com">wangchao</a>
 * @version 1.0
 * @since 2017-08-27 12:38:00
 */
@Service
public class ElasticsearchServiceImpl implements ElasticsearchService {
    private static final Logger logger = LoggerFactory.getLogger(ElasticsearchServiceImpl.class);

    @Resource
    private TransportClient transportClient;
    @Autowired
    private DataService dataService;

    @Override
    public void insertById(String index, String type, String id, Map<String, Object> dataMap) {
        createIndex(index, type);
        index = index + "_" + type;
        transportClient.prepareIndex(index, type, id).setSource(dataMap).get();
    }

    @Override
    public void batchInsertById(String index, String type, Map<String, Map<String, Object>> idDataMap) {
        createIndex(index, type);
        String indexType=  index + "_" + type;
        BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();

        idDataMap.forEach((id, dataMap) -> bulkRequestBuilder.add(transportClient.prepareIndex(indexType, type, id).setSource(JsonUtil.toJson(dataMap))));
        try {
            BulkResponse bulkResponse = bulkRequestBuilder.execute().get();
            if (bulkResponse.hasFailures()) {
                logger.error("elasticsearch批量插入错误, index=" + indexType + ", type=" + type + ", data=" + JsonUtil.toJson(idDataMap) + ", cause:" + bulkResponse.buildFailureMessage());
            }
        } catch (Exception e) {
            logger.error("elasticsearch批量插入错误, index=" + indexType + ", type=" + type + ", data=" + JsonUtil.toJson(idDataMap), e);
        }
    }

    @Override
    public void update(String index, String type, String id, Map<String, Object> dataMap) {
        this.insertById(index, type, id, dataMap);
    }

    @Override
    public void deleteById(String index, String type, String id) {
        index = index + "_" + type;
        transportClient.prepareDelete(index, type, id).get();
    }

    @Override
    public long count(String index, String type) throws Exception {
        index = index + "_" + type;
        SearchRequestBuilder search = transportClient.prepareSearch(index).setTypes(type);
        search.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        search.setSource(searchSourceBuilder).setFrom(0).setSize(0);
        SearchResponse response = search.execute().get();
        SearchHits hits = response.getHits();
        return hits.getTotalHits();
    }
    @Override
    public synchronized void createIndex(String index, String type){
        //改变索引
        index = index + "_" + type;
        //创建索引
        create(index, type,dataService.getEsMapping(type));
    }

    @Override
    public Boolean deleteIndex(String index, String type) {
        index = index + "_" + type;
        DeleteIndexResponse result = transportClient.admin().indices().prepareDelete(index).execute().actionGet();
        return result.isAcknowledged();
    }

    public synchronized void create(String index, String type, XContentBuilder xContentBuilder){
        if (!this.indexExists(index)) {
            IndicesAdminClient adminClient = this.getIndicesAdminClient();
            adminClient.prepareCreate(index).setSettings(getSettings()).addMapping(type, xContentBuilder).execute().actionGet();
        }
    }
    @Override
    public synchronized boolean indexExists(String index) {
        IndicesExistsRequest request = new IndicesExistsRequest(new String[]{index});
        IndicesExistsResponse response =this.getIndicesAdminClient().exists(request).actionGet();
        return response.isExists();
    }

    /**
     * 获取 IndicesAdminClient
     * @return IndicesAdminClient
     */
    public IndicesAdminClient getIndicesAdminClient() {
        return this.transportClient.admin().indices();
    }
    public static Settings.Builder getSettings() {
        return Settings.builder().put("index.number_of_shards", 5).put("index.number_of_replicas", 1).put("index.max_result_window", 50000000);
    }
}
